package com.demo.java.concurrent.threadpool;

import com.clearspring.analytics.util.Lists;
import com.demo.java.utils.LogUtils;
import com.demo.java.utils.ThreadUtils;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;

import java.util.List;
import java.util.Queue;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReferenceArray;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import static com.demo.java.utils.LogUtils.log;

/**
 * 添加转发任务
 * 消费转发任务
 * 处理转发任务
 * 	分割转发任务
 * 	添加转发子任务
 * 	消费转发子任务
 * 	处理转发子任务
 * 		查询子任务数据
 * 		查询到的数据转换
 * 		转换后的数据发送
 *
 * 	转发任务状态更新
 */
public class HistoryTransmitService {
    @Data
    @AllArgsConstructor
    static class ExportEntity {
        private int id;

    }
    private HistoryTransmitContext transmitContext;

    public HistoryTransmitService() {
        this.transmitContext = new HistoryTransmitContext(this);
    }

    // 添加一个转发任务
    public void addExportTask(ExportEntity exportEntity) {
        this.transmitContext.addExport(exportEntity);
    }


    static class HistoryTransmitContext {
        // 实际中应该使用自定义ThreadPoolExecutor方式创建
        private static final ExecutorService splitExportExecutor = Executors.newCachedThreadPool(ThreadUtils.createThreadFactory("split-executor-"));
        private static final ExecutorService loopExecutor = Executors.newSingleThreadScheduledExecutor(ThreadUtils.createThreadFactory("loop-executor-"));
        private static final ExecutorService runTaskExecutor = new ThreadPoolExecutor(0, 20, 60L, TimeUnit.SECONDS, new SynchronousQueue<>(), ThreadUtils.createThreadFactory("task-executor-"));
        private HistoryTransmitService historyTransmitService;
        private volatile List<TransmitHandler> exportHandlers = Lists.newArrayList();

        private final ReentrantLock contextLock = new ReentrantLock();
        private final Condition handlersEmpty = contextLock.newCondition();

        private AtomicBoolean shutdown = new AtomicBoolean(false);
        public HistoryTransmitContext(HistoryTransmitService historyTransmitService) {
            this.historyTransmitService = historyTransmitService;
            this.start();
        }

        private void start() {
            loopExecutor.execute(new Runnable() {
                public void run() {
                    try {
                        System.out.println(log("loop handlers start"));
                        while (!shutdown.get()) {
                            contextLock.lock();
                            try {
                                if (exportHandlers.size() == 0) {
                                    try {
                                        System.out.println(log("await"));
                                        handlersEmpty.await();
                                    } catch (InterruptedException e) {
                                        e.printStackTrace();
                                    }
                                }
                            } finally {
                                contextLock.unlock();
                            }

                            for (int i = 0; i < exportHandlers.size(); i++) {
                                TransmitHandler transmitHandler = exportHandlers.get(i);
                                Runnable runnable = transmitHandler.pollTask();
                                System.out.println(log("poll task, size: " + exportHandlers.size()));
                                if (runnable != null) {
                                    System.out.println(log("run task"));
                                    runTaskExecutor.execute(runnable);
                                }
                                System.out.println("-----------");
                            }

                            System.out.println(log("loop one end"));
                        }

                        shutdown();
                    } catch (Exception ex) {
                        ex.printStackTrace();
                    }
                }
            });
        }

        public synchronized void addExport(ExportEntity exportEntity) {
//            exportHandlers.offer(new TransmitHandler(exportEntity, this));
            System.out.println(log("add export"));

            exportHandlers.add(new TransmitHandler(exportEntity, this));

            if (exportHandlers.size() == 1) {
                contextLock.lock();
                try {
                    System.out.println(log("signal all"));
                    this.handlersEmpty.signalAll();
                } finally {
                    contextLock.unlock();
                }
            }
        }
        public synchronized void removeHandler(TransmitHandler transmitHandler) {
            System.out.println(log("remove handler"));
            this.exportHandlers.remove(transmitHandler);
            if (this.exportHandlers.size() == 0) {
                shutdown();
            }
        }

        public void executeSplitTask(ExportSplitRunner splitRunner) {
            System.out.println(log("run split task"));
            splitExportExecutor.execute(splitRunner);
        }

        public void shutdown() {
            if (!shutdown.getAndSet(true)) {
                System.out.println(log("shutdown"));
                splitExportExecutor.shutdown();
                loopExecutor.shutdown();
                runTaskExecutor.shutdown();
            }
        }

    }
    static class TransmitHandler {
        // 任务分割和任务转发
        private final Queue<Runnable> queue = new LinkedBlockingDeque<>();

        private int total = 2;
        private AtomicInteger successed = new AtomicInteger(0);
        private ExportEntity exportEntity;
        private HistoryTransmitContext transmitContext;

        public TransmitHandler(ExportEntity exportEntity, HistoryTransmitContext transmitContext) {
            this.exportEntity = exportEntity;
            this.transmitContext = transmitContext;
            this.transmitContext.executeSplitTask(new ExportSplitRunner(exportEntity, this));
        }

        public void startTransmit() {
//            System.out.println(LogUtils.log("transmit start"));
            int currentBatch = 0;
            int index = 0;
            /*while ((currentBatch = this.queryFromDB()) > 0) {
                for (int i = 0; i < currentBatch; i++) {
                    this.queue.offer(new TransmitRunner(this, index++));
                }
            }*/
            for (int i = 0; i < total; i++) {
                System.out.println(log("add transmit runner export: " + exportEntity.getId() + ", id: " + i));
                this.queue.offer(new TransmitRunner(this, i));
            }

            // 转发执行完毕
            System.out.println(log("export: " + exportEntity.getId() + ", task assign end, size: " + this.queue.size()));
        }

        private int exported = 0;
        private int queryFromDB() {
            // 批量从数据库中查询未执行的子任务
            if (exported < total) {
                if (exported + 10 <= total) {
                    exported += 10;
                    return 10;
                }
            }
            return total - exported;
        }
        public Runnable pollTask() {
            System.out.println(log("poll"));
            return this.queue.poll();
        }
        public void notifySuccessed() {
            System.out.println(log("notify success, export: " + exportEntity.getId()));
            if (successed.incrementAndGet() == total) {
                System.out.println("remove handler, export: " + exportEntity.getId());
                transmitContext.removeHandler(this);
            }
        }
        public int getExportId() {
            return this.exportEntity.getId();
        }
    }
    static class ExportSplitRunner implements Runnable {
        private ExportEntity exportEntity;
        private TransmitHandler transmitHandler;
        public ExportSplitRunner(ExportEntity exportEntity, TransmitHandler transmitHandler) {
            this.exportEntity = exportEntity;
            this.transmitHandler = transmitHandler;
        }
        public void run() {
//            System.out.println(log("export: " + transmitHandler.getExportId() + ", split task start running"));
            ThreadUtils.sleep(1);
            System.out.println(log("export: " + transmitHandler.getExportId() + ", split task end"));
            // 开始执行转发任务
            this.transmitHandler.startTransmit();
        }
    }
    static class TransmitRunner implements Runnable {
        private TransmitHandler transmitHandler;
        private int index;
        public TransmitRunner(TransmitHandler transmitHandler, int index) {
            this.transmitHandler = transmitHandler;
            this.index = index;
        }
        public void run() {
            System.out.println(log("export: " + transmitHandler.getExportId() + ", transmit runner start, index: " + this.index));
            Sender.send(transmitHandler.getExportId(), Fetcher.fetch());
            System.out.println(log("export: " + transmitHandler.getExportId() + ", transmit runner end, index: " + this.index));

            // 通知handlerContext当前子task处理完成
            this.transmitHandler.notifySuccessed();
        }
    }
    static class Fetcher {
        public static int fetch() {
            return (int)(Math.random()*10);
        }
    }
    static class Sender {
        public static void send(int exportId, int data) {
            System.out.println(log("export: " + exportId + ", run send data start, data: " + data));
            ThreadUtils.sleep(1);
            System.out.println(log("export: " + exportId + ", run send data end, data: " + data));
        }

    }

    public static void main(String[] args) {
        HistoryTransmitService service = new HistoryTransmitService();

        for (int i = 0; i < 2; i++) {
            service.addExportTask(new ExportEntity(i));
        }
    }

}
